%%% -------------------------------------------------------------------
%%% Author  : Administrator
%%% Description :
%%%
%%% Created : 2009-3-23
%%% -------------------------------------------------------------------
-module(node_writers).

-behaviour(gen_server).
%% --------------------------------------------------------------------
%% Include files
%% --------------------------------------------------------------------
-include("ring.hrl").
-include("map.hrl").

%% --------------------------------------------------------------------
%% External exports
-export([start_link/1]).
-export([get_current_writer/1, update_nodewriters/1, report_state/0]).

%% gen_server callbacks
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).

-record(state, {current_idx = 1}).
-record(node_writers, {
    node_id :: node_id(),
    writers = [] :: [pid()]
}).

%% ====================================================================
%% External functions
%% ====================================================================
start_link(RingNodes) ->
    gen_server:start_link({local, ?MODULE}, ?MODULE, RingNodes, []).

%% ====================================================================
%% Server functions
%% ====================================================================
get_current_writer(RingNodeId) ->
    gen_server:call(?MODULE, {get_current_writer_for, RingNodeId}).

update_nodewriters(RingNodes) ->
    gen_server:call(?MODULE, {update_ringnodes, RingNodes}).

report_state() ->
    gen_server:call(?MODULE, {report_state}).

%% --------------------------------------------------------------------
%% Function: init/1
%% Description: Initiates the server
%% Returns: {ok, State}          |
%%          {ok, State, Timeout} |
%%          ignore               |
%%          {stop, Reason}
%% --------------------------------------------------------------------
init(RingNodes) ->
    log4erl:debug("starting node_writer for: ~p", [RingNodes]),
    process_flag(trap_exit, true),

    ets:new(node_writers, [named_table, ordered_set, private, {keypos, #node_writers.node_id}]),
    ets:new(writer_to_node, [named_table, ordered_set, private]),
    
    new_nodewriter(RingNodes),
    
    {ok, #state{}}.

%% --------------------------------------------------------------------
%% Function: handle_call/3
%% Description: Handling call messages
%% Returns: {reply, Reply, State}          |
%%          {reply, Reply, State, Timeout} |
%%          {noreply, State}               |
%%          {noreply, State, Timeout}      |
%%          {stop, Reason, Reply, State}   | (terminate/2 is called)
%%          {stop, Reason, State}            (terminate/2 is called)
%% --------------------------------------------------------------------
handle_call({report_state}, _From, #state{current_idx=Idx}=State) ->
    io:format("node_writers: ~n", []),
    lists:foreach(fun(#node_writers{node_id=NodeId, writers=WriterPids}) ->
        io:format("node_writer_for: ~p, idx: ~p~n", [NodeId, Idx]),
        lists:foreach(fun(Pid) ->
            io:format("\t~p(~p)", [Pid, erlang:is_process_alive(Pid)])
        end, WriterPids),
        io:format("~n")
    end, ets:tab2list(node_writers)),
    {reply, ok, State};
handle_call({update_ringnodes, RingNodes}, _From, State) ->
    terminate_nodewriters(),
    new_nodewriter(RingNodes),
    {reply,ok,State};
handle_call({get_current_writer_for, RingNodeId}, _From, #state{current_idx=Idx}=State) ->
    [#node_writers{writers=WriterPids}] = ets:lookup(node_writers, RingNodeId),
    NewIdx = (Idx rem length(WriterPids)) + 1,
    Writer = lists:nth(NewIdx, WriterPids),
%    ets:insert(node_writers, NodeWriters#node_writers{current_idx=NewIdx}),
    {reply, Writer, State#state{current_idx=NewIdx}}.

%% --------------------------------------------------------------------
%% Function: handle_cast/2
%% Description: Handling cast messages
%% Returns: {noreply, State}          |
%%          {noreply, State, Timeout} |
%%          {stop, Reason, State}            (terminate/2 is called)
%% --------------------------------------------------------------------
handle_cast(_Msg, State) ->
    {noreply, State}.

%% --------------------------------------------------------------------
%% Function: handle_info/2
%% Description: Handling all non call/cast messages
%% Returns: {noreply, State}          |
%%          {noreply, State, Timeout} |
%%          {stop, Reason, State}            (terminate/2 is called)
%% --------------------------------------------------------------------
handle_info({'EXIT', WriterPid, Reason}, State) ->
    log4erl:debug("some writer(~p) exit with: ~p ", [WriterPid, Reason]),
    case ets:lookup(writer_to_node, WriterPid) of
        [{WriterPid, RingNodeId}] -> 
            log4erl:info("Writer(~p) for node(~p) exit with: ~p~n", [WriterPid, RingNodeId, Reason]),
            [#node_writers{writers=WriterPids}=NodeWriters] = ets:lookup(node_writers, RingNodeId),
            NewWriterPids = lists:dropwhile(fun(Pid) -> Pid == WriterPid end, WriterPids),
            ets:delete(writer_to_node, WriterPid),
            ets:insert(node_writers, NodeWriters#node_writers{writers=NewWriterPids}),
            {noreply, #state{current_idx=1}=State};
        [] ->
            {noreply, State}
    end;

handle_info(_Info, State) ->
    log4erl:warn("some writer conn may break: ~p", [_Info]),
    {noreply, State}.

%% --------------------------------------------------------------------
%% Function: terminate/2
%% Description: Shutdown the server
%% Returns: any (ignored by gen_server)
%% --------------------------------------------------------------------
terminate(_Reason, _State) ->
    log4erl:info("node_writer terminated: ~p", [_Reason]),
    terminate_nodewriters(),
    ok.

clear_nodewriters(RingNodeId) ->
    case ets:lookup(node_writers, RingNodeId) of
        [#node_writers{writers = WriterPids}] ->
            lists:foreach(fun(Pid) ->
                case erlang:is_process_alive(Pid) of
                    true ->
                        Pid ! {stop};
                    false ->
                        stopped
                end,
                ets:delete(writer_to_node, Pid)
            end, WriterPids),
            ets:delete(node_writers, RingNodeId);
        [] ->
            deleted
    end.

terminate_nodewriters() ->
    NodewriterList = ets:tab2list(node_writers),
    lists:foreach(fun(#node_writers{writers = OldNodeWriters}) -> 
        lists:foreach(fun(OldNodeWriter) ->
            case erlang:is_process_alive(OldNodeWriter) of
                true ->
                    OldNodeWriter ! {stop};
                false ->
                    stopped
            end                            
        end, OldNodeWriters)
    end, NodewriterList),
    ets:delete_all_objects(writer_to_node),
    ets:delete_all_objects(node_writers).

%% --------------------------------------------------------------------
%% Func: code_change/3
%% Purpose: Convert process state when code is changed
%% Returns: {ok, NewState}
%% --------------------------------------------------------------------
code_change(_OldVsn, State, _Extra) ->
    {ok, State}.

%% --------------------------------------------------------------------
%%% Internal functions
%% --------------------------------------------------------------------
new_nodewriter(RingNodes) ->
    NodeWriters = lists:map(fun create_writers_for/1, RingNodes),
    lists:foreach(fun({NodeId, WriterPids}) ->
        ets:insert(node_writers , #node_writers{node_id=NodeId, writers=WriterPids}),
        lists:foreach(fun(WriterPid) ->
            ets:insert(writer_to_node, {WriterPid, NodeId})
        end, WriterPids)
    end, NodeWriters).

create_writers_for(#ring_node{id=NodeId, ip=Ip, access_port=Port}) ->
    Writers = [new_writer(Ip, Port) || _ <- lists:seq(1, ?WRITER_POOL_SIZE)],
    {NodeId, Writers}.

new_writer(Address, Port) ->
    Pid = spawn_link(fun() ->
        {ok, Socket} = gen_tcp:connect(Address, Port, [binary, {packet, 4}, {active,once}]),
        write_loop(Socket)
    end),
    Pid.
    
write_loop(Socket) ->
    receive
        {put, Key, #cell{}=Cell, From} ->
            gen_tcp:send(Socket, term_to_binary({put, Key, Cell})),
            receive
                {tcp, Socket, Bin} ->
                    Key = binary_to_term(Bin),
                    gen_server:reply(From, Key),
                    log4erl:info("write to socket success, key=~p", [Key]),
                    inet:setopts(Socket, [{active, once}]),
                    write_loop(Socket);
                {tcp_closed, Socket} ->
                    log4erl:warn("Socket closed"),
                    exit(closed);
                {tcp_error, Socket, Reason} ->
                    exit(Reason)
            end;
        {stop} ->
            stop
    end.
